package MRRestart.STjoin;


import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import yeliuhuishi.File.DelAllFile;

import java.io.File;
import java.io.IOException;
import java.util.StringTokenizer;

/**
 * 单表关联
 * 描述 child-parent（孩子父母），输出grandchild-grandparent（孙子--爷爷）表
 */
public class STjoin {
    public static int time = 0;

    public static void main(String[] args) throws IOException, ClassNotFoundException, InterruptedException {
        String input = "src/main/java/MRRestart/STjoin/data/input";
        String output = "src/main/java/MRRestart/STjoin/data/output";

        DelAllFile.delAllFile(new File(output));

        Configuration conf = new Configuration();
        Job job = Job.getInstance(conf, "mr");

        job.setJarByClass(STjoin.class);
        job.setMapperClass(Map.class);
        job.setReducerClass(Reduce.class);


        job.setOutputKeyClass(Text.class);
        job.setOutputValueClass(Text.class);

        FileInputFormat.setInputPaths(job, new Path(input));

        FileOutputFormat.setOutputPath(job, new Path(output));

        System.out.println(job.waitForCompletion(true) ? "成功" : "失败");
    }

    /**
     * map将输出分割child和parent，然后正序输出一次作为右表，
     * 反序输出一次作为左表，需要注意的是在输出的value中必须
     * 加上左右表的区别标识
     */
    public static class Map extends Mapper<Object, Text, Text, Text> {
        @Override
        protected void
        map(Object key, Text value, Context context) throws IOException, InterruptedException {
            String childnam;// 孩子名称
            String parentnam;//父母名称
            String relationtype;//左右表标识
            //  输入的一行预处理文本
            StringTokenizer itr = new StringTokenizer(value.toString());
            String[] values = new String[2];
            int i = 0;
            while (itr.hasMoreTokens()) {
                values[i] = itr.nextToken();
                i++;
            }
            if (values[0].compareTo("child") != 0) {
                childnam = values[0];
                parentnam = values[1];

                //  输出左表
                relationtype = "1";
                context.write(new Text(values[1]), new Text(relationtype + "+" + childnam + "+" + parentnam));

                //  输出右表
                relationtype = "2";
                context.write(new Text(values[0]), new Text(relationtype + "+" + childnam + "+" + parentnam));
            }
        }
    }

    public static class Reduce extends Reducer<Text, Text, Text, Text> {
        @Override
        protected void reduce(Text key, Iterable<Text> values, Context context) throws IOException, InterruptedException {
            if (0 == time) {
                context.write(new Text("grandchild"), new Text("grandparent"));
                time++;
            }

            int grandchildnum = 0;
            String[] grandchild = new String[10];
            int grandparentnum = 0;
            String[] grandparent = new String[10];

            for (Text value : values) {
                String record = value.toString();
                int len = record.length();
                int i = 2;
                if (0 == len) {
                    continue;
                }

                // 取得左右表标识
                char relationtype = record.charAt(0);

                // 定义孩子和父母变量
                StringBuilder childnam = new StringBuilder();
                StringBuilder parentname = new StringBuilder();
                // 获得value-list中value的child
                while (record.charAt(i) != '+') {
                    childnam.append(record.charAt(i));
                    i++;
                }
                i++;

                // 获得value-list中value的parent
                while (i < len) {
                    parentname.append(record.charAt(i));
                    i++;
                }
                // 左表，取出child放入grandchildren
                if ('1' == relationtype) {
                    grandchild[grandchildnum] = childnam.toString();
                    grandchildnum++;
                }

                // 右表，取出parent放入grandparent
                if ('2' == relationtype) {
                    grandparent[grandparentnum] = parentname.toString();
                    grandparentnum++;
                }
            }

            // grandchild和grandparent数组求笛卡尔积
            if (0 != grandchildnum && 0 != grandparentnum) {
                for (int m = 0; m < grandchildnum; m++) {
                    for (int n = 0; n < grandparentnum; n++) {
                        context.write(new Text(grandchild[m]), new Text(grandparent[n]));
                    }
                }
            }

        }


    }
}
